package cn.lgwen.candy.connector.kafka010.consumer;

import cn.lgwen.candy.connector.kafka010.source.KafkaSource010;
import cn.lgwen.candy.runtime.driver.SourceDriver;
import cn.lgwen.candy.runtime.queue.MemoryBlockQueue;
import cn.lgwen.candy.runtime.queue.Queue;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.junit.Before;
import org.junit.Test;

import java.util.*;

/**
 * 2019/12/26
 * aven.wu
 * danxieai258@163.com
 */
public class KafkaOffsetManager {

    Properties properties;

    @Before
    public void setProperties() {
        properties = new Properties();
        properties.put("bootstrap.servers","10.20.128.210:19091");
        properties.put("group.id","group_autoManageOffset");
        properties.put("auto.offset.reset","earliest");
        properties.put("enable.auto.commit","false");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    }

    @Test
    public void autoManageOffset2(){
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList("bee.dwd.flow"));
        while(true){
            ConsumerRecords<String, String> records = consumer.poll(1000);
            //通过records获取这个集合中的数据属于那几个partition
            Set<TopicPartition> partitions = records.partitions();
            for(TopicPartition tp : partitions){
                //通过具体的partition把该partition中的数据拿出来消费
                List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
                for(ConsumerRecord<String, String> r : partitionRecords){
                    System.out.println(r.offset()   +"     "+r.key()+"     "+r.value());
                }
                //获取新这个partition中的最后一条记录的offset并加1 那么这个位置就是下一次要提交的offset
                long newOffset = partitionRecords.get(partitionRecords.size() - 1).offset() + 1;
                consumer.commitSync(Collections.singletonMap(tp,new OffsetAndMetadata(newOffset)));
            }
        }
    }

    @Test
    public void testKafkaSourceFetcher() {
        KafkaSource010 kafkaSource010 = new KafkaSource010(properties, "bee.dwd.flow");
        Queue<String> queue = new MemoryBlockQueue<>();
        SourceDriver<String > stringSourceDriver = new SourceDriver<>(kafkaSource010);
        stringSourceDriver.registerForward(queue);
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    System.out.println(queue.take());
                }
            }
        });
        thread.start();
        stringSourceDriver.run();
    }

}
